package mq

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"

	"e-commerce/services/inventory-service/internal/repository"
	orderpb "e-commerce/services/order-service/api/pb"

	servicetoken "e-commerce/services/auth-service/api/servietoken"

	"github.com/streadway/amqp"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/otel"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"gopkg.in/yaml.v3"
)

type RabbitMQ struct {
	conn        *amqp.Connection
	ch          *amqp.Channel
	orderClient *orderClient
}

type mqconfig struct {
	Mq MqConfig `yaml:"rabbitmq"`
}

type MqConfig struct {
	Host     string `yaml:"host"`
	Port     string `yaml:"port"`
	User     string `yaml:"user"`
	Password string `yaml:"password"`
	Vhost    string `yaml:"vhost"`
}

func NewRabbitMQ(configPath string) *RabbitMQ {
	yamlFile, err := os.ReadFile(configPath)
	if err != nil {
		log.Fatalf("读取配置文件失败: %v", err)
	}

	var cfg mqconfig
	err = yaml.Unmarshal(yamlFile, &cfg)
	if err != nil {
		log.Fatalf("解析配置文件失败: %v", err)
	}

	conn, err := amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%s/%s", cfg.Mq.User, cfg.Mq.Password, cfg.Mq.Host, cfg.Mq.Port, cfg.Mq.Vhost))
	if err != nil {
		log.Fatalf("连接RabbitMQ失败: %v", err)
	}
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("创建通道失败: %v", err)
	}
	orderClient, err := NewOrderClient(configPath)
	if err != nil {
		log.Fatalf("创建order客户端失败: %v", err)
	}
	return &RabbitMQ{conn: conn, ch: ch, orderClient: orderClient}
}

func (mq *RabbitMQ) NewOrderQueue() error {
	//创建带死信队列的订单队列inventory.reserve
	_, err := mq.ch.QueueDeclare("inventory.reserve", true, false, false, false, amqp.Table{
		"x-dead-letter-exchange":    "inventory-reserve-dlx",
		"x-message-ttl":             10000,
		"x-dead-letter-routing-key": "inventory-reserve-key",
	})
	if err != nil {
		return fmt.Errorf("创建订单队列失败: %v", err)
	}
	//创建死信交换机
	err = mq.ch.ExchangeDeclare("inventory-reserve-dlx", "direct", true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("创建死信交换机inventory-reserve-dlx失败: %v", err)
	}
	//创建死信队列
	_, err = mq.ch.QueueDeclare("inventory-reserve-dlq", true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("创建死信队列inventory-reserve-dlq失败: %v", err)
	}
	//绑定死信队列
	err = mq.ch.QueueBind("inventory-reserve-dlq", "inventory-reserve-key", "inventory-reserve-dlx", false, nil)
	if err != nil {
		return fmt.Errorf("绑定死信队列inventory-reserve-dlq失败: %v", err)
	}

	//创建带死信队列的订单队列inventory.rollback
	_, err = mq.ch.QueueDeclare("inventory.rollback", true, false, false, false, amqp.Table{
		"x-dead-letter-exchange":    "inventory-rollback-dlx",
		"x-message-ttl":             10000,
		"x-dead-letter-routing-key": "inventory-rollback-key",
	})
	if err != nil {
		return fmt.Errorf("创建订单队列inventory.rollback失败: %v", err)
	}
	//创建死信交换机
	err = mq.ch.ExchangeDeclare("inventory-rollback-dlx", "direct", true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("创建死信交换机inventory-rollback-dlx失败: %v", err)
	}
	//创建死信队列
	_, err = mq.ch.QueueDeclare("inventory-rollback-dlq", true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("创建死信队列inventory-rollback-dlq失败: %v", err)
	}
	//绑定死信队列
	err = mq.ch.QueueBind("inventory-rollback-dlq", "inventory-rollback-key", "inventory-rollback-dlx", false, nil)
	if err != nil {
		return fmt.Errorf("绑定死信队列inventory-rollback-dlq失败: %v", err)
	}

	//创建交换机
	err = mq.ch.ExchangeDeclare("order.exchange", "direct", true, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("创建交换机order.exchange失败: %v", err)
	}
	//绑定交换机
	err = mq.ch.QueueBind("inventory.reserve", "order.created", "order.exchange", false, nil)
	if err != nil {
		return fmt.Errorf("绑定交换机inventory.reserve失败: %v", err)
	}
	err = mq.ch.QueueBind("inventory.rollback", "order.canceled", "order.exchange", false, nil)
	if err != nil {
		return fmt.Errorf("绑定交换机inventory.rollback失败: %v", err)
	}
	err = mq.ch.QueueBind("inventory.rollback", "order.failed", "order.exchange", false, nil)
	if err != nil {
		return fmt.Errorf("绑定交换机inventory.rollback失败: %v", err)
	}
	return nil
}

func (mq *RabbitMQ) Publish(routingKey string, message []byte) error {
	for i := 0; i < 3; i++ {
		err := mq.ch.Publish("order.exchange", routingKey, false, false, amqp.Publishing{
			ContentType: "application/json",
			Body:        message,
		})
		if err == nil {
			return nil
		}
		log.Printf("发送消息失败，重试第%d次,错误信息: %v", i+1, err)
	}
	return fmt.Errorf("发送消息失败")
}

func (mq *RabbitMQ) ReservedPublish(message []byte) error {
	return mq.Publish("stock.reserved", message)
}

func (mq *RabbitMQ) FailedPublish(message []byte) error {
	return mq.Publish("stock.failed", message)
}

func (mq *RabbitMQ) Consume(
	queue string,
	handler func(msg amqp.Delivery, inventoryRepository *repository.InventoryRepository, mq *RabbitMQ) error,
	inventoryRepository *repository.InventoryRepository,
) error {
	msgs, err := mq.ch.Consume(queue, "", false, false, false, false, nil)
	if err != nil {
		return fmt.Errorf("队列%s消费消息失败: %v", queue, err)
	}
	go func() {
		for msg := range msgs {
			err := handler(msg, inventoryRepository, mq)
			if err != nil {
				log.Printf("处理消息失败: %v,重试次数: %v", err, msg.Headers["x-retries"])
				if err := Republish(mq, msg); err != nil {
					log.Printf("重试消息失败: %v", err)
				}
			} else {
				log.Printf("处理消息成功，确认消息 %s", string(msg.Body))
				msg.Ack(false)
			}
		}
	}()
	return nil
}

func HandleReservedMessage(msg amqp.Delivery, inventoryRepository *repository.InventoryRepository, mq *RabbitMQ) error {
	var reservedMessage ReversedMessage
	var stockMessage StockMessage
	err := json.Unmarshal(msg.Body, &reservedMessage)
	if err != nil {
		return fmt.Errorf("解析预扣库存消息失败: %v", err)
	}
	stockMessage.OrderID = reservedMessage.OrderID
	stockMessage.ProductID = reservedMessage.ProductID
	err = inventoryRepository.ReserveStock(reservedMessage.ProductID, reservedMessage.ReversedNum)
	if err != nil {
		log.Printf("预扣库存失败: %v", err)
		stockMessage.ReserveResp = false
		stockMessageBytes, err := json.Marshal(stockMessage)
		if err != nil {
			return fmt.Errorf("序列化库存预留成功消息失败: %v", err)
		}
		mq.FailedPublish(stockMessageBytes)
		return fmt.Errorf("预扣库存失败: %v", err)
	}
	log.Printf("预扣库存成功")

	stockMessage.ReserveResp = true
	log.Printf("stockMessage: %v", stockMessage)
	stockMessageBytes, err := json.Marshal(stockMessage)
	if err != nil {
		return fmt.Errorf("序列化库存预留成功消息失败: %v", err)
	}
	log.Printf("序列化库存预留成功消息成功")
	err = mq.ReservedPublish(stockMessageBytes)
	if err != nil {
		log.Printf("发送库存预留成功消息失败: %v", err)
		return fmt.Errorf("发送库存预留成功消息失败: %v", err)
	}
	log.Printf("发送库存预留成功消息成功")
	return nil
}

func HandleRollbackMessage(msg amqp.Delivery, inventoryRepository *repository.InventoryRepository, mq *RabbitMQ) error {
	var rollbackMessage StockRollbackMessage
	err := json.Unmarshal(msg.Body, &rollbackMessage)
	if err != nil {
		return fmt.Errorf("解析回滚库存消息失败: %v", err)
	}
	log.Printf("rollbackMessage: %v", rollbackMessage)
	//手动生成span
	tracer := otel.Tracer("inventory-service")
	ctx, span := tracer.Start(context.Background(), "HandleRollbackMessage")
	defer span.End()
	productID, quantity, err := mq.orderClient.GetOrderData(ctx, rollbackMessage.OrderID)
	if err != nil {
		return fmt.Errorf("获取订单数据失败: %v", err)
	}
	err = inventoryRepository.ReleaseStock(productID, quantity)
	if err != nil {
		return fmt.Errorf("回滚库存失败: %v", err)
	}
	return nil
}

// 生成order服务的grpc客户端
type Config struct {
	OrderService order_service `yaml:"order_service"`
}

type order_service struct {
	Addr string `yaml:"addr"`
}

type orderClient struct {
	client orderpb.OrderServiceClient
	conn   *grpc.ClientConn
}

func NewOrderClient(configPath string) (*orderClient, error) {
	yamlFile, err := os.ReadFile(configPath)
	if err != nil {
		return nil, err
	}
	var config Config
	err = yaml.Unmarshal(yamlFile, &config)
	if err != nil {
		return nil, err
	}
	conn, err := grpc.NewClient(
		config.OrderService.Addr,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
	)
	if err != nil {
		return nil, err
	}
	return &orderClient{
		client: orderpb.NewOrderServiceClient(conn),
		conn:   conn,
	}, nil
}

func (c *orderClient) GetOrderData(ctx context.Context, orderID int64) (string, int32, error) {
	//往context添加token
	ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+servicetoken.ServiceTokens["inventory-service"])
	orderresp, err := c.client.GetOrder(ctx, &orderpb.GetOrderRequest{OrderId: orderID})
	if err != nil {
		return "", 0, err
	}
	return orderresp.Order.ProductId, orderresp.Order.Quantity, nil
}

func (mq *RabbitMQ) GetOrderData(ctx context.Context, orderID int64) (string, int32, error) {
	//手动生成span
	tracer := otel.Tracer("inventory-service")
	ctx, span := tracer.Start(ctx, "GetOrderData")
	defer span.End()
	return mq.orderClient.GetOrderData(ctx, orderID)
}

func Republish(mq *RabbitMQ, msg amqp.Delivery) error {
	if msg.Headers["x-retries"].(int32) >= 2 {
		msg.Nack(false, false)
		return fmt.Errorf("重试次数超过3次")
	}
	msg.Headers["x-retries"] = msg.Headers["x-retries"].(int32) + 1
	err := mq.ch.Publish("order.exchange", msg.RoutingKey, false, false, amqp.Publishing{
		ContentType: "application/json",
		Body:        msg.Body,
		Headers:     msg.Headers,
	})
	if err != nil {
		msg.Nack(false, true)
		return fmt.Errorf("重新发送回滚消息失败: %v,重试次数: %v", err, msg.Headers["x-retries"])
	}
	msg.Ack(false)
	return nil
}
